package com.atguigu.day05;

import com.atguigu.bean.OrderEvent;
import com.atguigu.bean.TxEvent;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
import org.apache.flink.util.Collector;

import java.util.HashMap;

public class Flink01_Project_Order {
    public static void main(String[] args) throws Exception {
        //1.获取流的执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        //2.获取订单数据以及交易数据并转为JavaBean
        SingleOutputStreamOperator<OrderEvent> orderEventStream = env.readTextFile("input/OrderLog.csv")
                .map(new MapFunction<String, OrderEvent>() {
                    @Override
                    public OrderEvent map(String value) throws Exception {
                        String[] split = value.split(",");
                        return new OrderEvent(Long.parseLong(split[0]), split[1], split[2], Long.parseLong(split[3]));
                    }
                });
        
        //交易数据
        SingleOutputStreamOperator<TxEvent> txEventStream = env.readTextFile("input/ReceiptLog.csv")
                .map(new MapFunction<String, TxEvent>() {
                    @Override
                    public TxEvent map(String value) throws Exception {
                        String[] split = value.split(",");
                        return new TxEvent(split[0], split[1], Long.parseLong(split[2]));
                    }
                });

        //3.将两条流链接在一起
        ConnectedStreams<OrderEvent, TxEvent> connect = orderEventStream.connect(txEventStream);

        //TODO 4.将相同交易码的数据聚合到一块  （多并行度的情况下要指定keyby，在connect之前之后都可以）
        ConnectedStreams<OrderEvent, TxEvent> orderEventTxEventConnectedStreams = connect.keyBy("txId", "txId");

        orderEventTxEventConnectedStreams.process(new KeyedCoProcessFunction<String, OrderEvent, TxEvent, String>() {

            //创建两个集合分别存订单数据和交易数据
            private HashMap<String, OrderEvent> orderMap = new HashMap<>();
            private HashMap<String, TxEvent> txMap = new HashMap<>();

            //订单数据
            @Override
            public void processElement1(OrderEvent value, Context ctx, Collector<String> out) throws Exception {

                //1.先看交易数据集合中是否有能关联上的数据
                if (txMap.containsKey(value.getTxId())) {
                    //有能关联上的数据
                    out.collect("订单：" + value.getOrderId() + "对账成功！！！！！！");
                    //对应已经成功关联的数据，需要在集合中将这些数据删除，防止OOM
                    txMap.remove(value.getTxId());
                } else {
                    //没有能关联上的数据，则把自己存入Map集合中
                    orderMap.put(value.getTxId(), value);
                }

            }

            //交易数据
            @Override
            public void processElement2(TxEvent value, Context ctx, Collector<String> out) throws Exception {

                //1.先看订单数据集合中是否有能关联上的数据
                if (orderMap.containsKey(value.getTxId())) {
                    //有能关联上的数据
                    out.collect("订单：" + orderMap.get(value.getTxId()).getOrderId() + "对账成功！！！！！！");
                    //对应已经成功关联的数据，需要在集合中将这些数据删除，防止OOM
                    orderMap.remove(value.getTxId());
                } else {
                    //没有能关联上的数据，则把自己存入Map集合中
                    txMap.put(value.getTxId(), value);
                }
            }
        }).print();


        env.execute();


    }
}
